package com.cbh.amadeus.client.utils;

import com.cbh.amadeus.client.config.SubscriberConfig;
import com.cbh.amadeus.client.domain.MessageRepositoryModel;
import com.cbh.amadeus.client.factory.MessageResendFactory;
import com.cbh.amadeus.client.receiver.tool.MessageRepositoryModelTool;
import com.cbh.amadeus.common.enums.MessageApproach;
import com.cbh.amadeus.common.enums.MessageType;
import com.cbh.amadeus.common.json.Json;
import com.cbh.amadeus.common.protocol.AmadeusMessage;
import com.cbh.amadeus.common.protocol.AmadeusMessageProtocol;
import com.cbh.amadeus.common.protocol.AmadeusMessageWrap;
import com.cbh.amadeus.common.utils.StringPool;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/**
 * @author ChenBiHui
 * @date 2025/5/8
 */
@Slf4j
public class AmadeusMessageTemplate {

    @Resource
    private SubscriberConfig config;
    @Resource
    private MessageRepositoryModelTool tool;

    public <T> void direct(String tenant, String topic, T message) {
        compose(config, MessageApproach.DIRECT, tenant, topic, message, 0L, TimeUnit.MILLISECONDS);
    }

    public <T> void delay(String tenant, String topic, T message, long duration) {
        delay(tenant, topic, message, duration, TimeUnit.SECONDS);
    }

    public <T> void delay(String tenant, String topic, T message, long duration, TimeUnit unit) {
        compose(config, MessageApproach.DELAY, tenant, topic, message, duration, unit);
    }

    public <T> void compose(SubscriberConfig config, MessageApproach approach, String tenant, String topic, T message, long duration, TimeUnit unit) {
        MessageRepositoryModel model = model(config, approach, tenant, topic, message, duration, unit);
        tool.commit(model);
        log.info("===== [{}] Topic [{}] has been commit with [{}] =====", approach.name(), topic, model.getMessageId());
        MessageResendFactory.call(tenant);
        Optional.ofNullable(ChannelReference.get())
                .ifPresent(channel -> channel.writeAndFlush(amadeus(config, model)));
    }

    private <T> MessageRepositoryModel model(SubscriberConfig config, MessageApproach approach, String tenant, String topic, T message, long duration, TimeUnit unit) {
        MessageRepositoryModel model = new MessageRepositoryModel();
        model.setId(tool.sequence());
        model.setApproach(approach.getType().byteValue());
        model.setAchieve(System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(duration, unit));
        model.setTenant(tenant);
        model.setProvider(config.getSubscriberName());
        model.setTopic(topic);
        model.setMessageId(messageId());
        model.setPayload(Json.payloadJson(message));
        model.setStatus(MessageRepositoryModel.MessageRepositoryEnum.ACTIVE.name());
        model.setReject(0);
        model.setCreateTime(LocalDateTime.now());
        model.setUpdateTime(LocalDateTime.now());
        return model;
    }

    private AmadeusMessageWrap amadeus(SubscriberConfig config, MessageRepositoryModel model) {
        AmadeusMessage amadeus = new AmadeusMessage();
        amadeus.setTenant(model.getTenant());
        amadeus.setProvider(config.getSubscriberName());
        amadeus.setTopic(model.getTopic());
        amadeus.setMessageId(model.getMessageId());
        amadeus.setPayload(model.getPayload());
        AmadeusMessageWrap wrap = new AmadeusMessageWrap();
        wrap.setMagic(AmadeusMessageProtocol.MAGIC);
        wrap.setType(MessageType.GEN.getType().byteValue());
        wrap.setApproach(model.getApproach());
        wrap.setAchieve(model.getAchieve());
        wrap.setMessage(amadeus);
        return wrap;
    }

    private String messageId() {
        ThreadLocalRandom random = ThreadLocalRandom.current();
        return new UUID(random.nextLong(), random.nextLong()).toString().replace(StringPool.DASH, StringPool.EMPTY);
    }
}
